package com.shujia.dw

import com.shujia.bean.CaseClass.MergeLocation
import com.shujia.util.SparkTool
import com.shujia.util.Config
import org.apache.spark.sql.{SQLContext, SaveMode}


object MakeMergeLocaltionApp extends SparkTool {


  /**
    * 编写spark业务逻辑
    *
    */
  override def run(args: Array[String]): Unit = {
    /**
      * 构建数据融合表
      *
      * 读取四类数据
      * ddr  dpi  oidd wcdr
      *
      */

    val ddrpath = Config.getString("ddr.path") + "day_id=" + day_id
    val oiddpath = Config.getString("oidd.path") + "day_id=" + day_id
    val dpipath = Config.getString("dpi.path") + "day_id=" + day_id
    val wcdrpath = Config.getString("wcdr.path") + "day_id=" + day_id

    val mergeLocationPath = Config.getString("mergelocation.path") + "day_id=" + day_id

    LOGGER.info("ddr:" + ddrpath)
    LOGGER.info("oidd:" + oiddpath)
    LOGGER.info("wcdr:" + wcdrpath)
    LOGGER.info("dpi:" + dpipath)

    LOGGER.info("融合表输出目录：" + mergeLocationPath)

    val ddrRDD = sc.textFile(ddrpath)
    val dpiRDD = sc.textFile(dpipath)
    val oiddRDD = sc.textFile(oiddpath)
    val wcdrRDD = sc.textFile(wcdrpath)


    //合并四类数据
    val mergeRDD = ddrRDD.union(dpiRDD).union(oiddRDD).union(wcdrRDD)


    //删除输出路径
    /* val path = new Path(mergeLocationPath)
     val fileSystem = FileSystem.newInstance(new Configuration)
     if (fileSystem.exists(path)) {
       fileSystem.delete(path, true)
     }*/


    val sqlContext = new SQLContext(sc)

    //导入隐私转换
    import sqlContext.implicits._

    //将rdd转换成df
    val mergerDF = mergeRDD
      .coalesce(10)
      .map(line => {
        val split = line.split("\t")
        MergeLocation(split(0), split(1), split(2), split(3), split(4), split(5), split(6), split(7), split(8), split(9))
      }).toDF()

    //将数据写入hdfs
    //parquet 数据压缩，数据自带列的描述
    mergerDF
      .write
      .mode(SaveMode.Overwrite)
      .parquet(mergeLocationPath)


  }



  /**
    * 初始化方法，在子类设置spark运行时需要的参数
    * conf.setMaster("local")
    */
  override def init(): Unit = {
    //conf.setMaster("local[8]")
  }
}
